package com.atguigu.realtime.app

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.Constansts
import com.atguigu.realtime.app.GMVApp.streamingContext
import com.atguigu.realtime.utils.{MyKafkaUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.InputDStream
import redis.clients.jedis.Jedis

/**
 * Created by Smexy on 2021/7/10
 *
 *  {"birthday":"2007-04-30","login_name":"ZbkVWXYbjRHKDOhtylII","gender":"F","create_time":"2021-07-07 14:09:47",
 *  "passwd":"pwdnLONydXCTLbXIxmVPnWT","nick_name":"qDWYcsESCigkJCPtzsmk",
 *  "head_img":"http://NmXZgIVCdHdvRXloLONvNJPgkJNQMndDHBINOHSs",
 *  "name":"CgaHkfWrRlFsXKJpwmVQttQWkmFnyF","user_level":"2",
 *  "phone_num":"13510772362","id":"2","email":"ijvBxFMR@XoA.com"}
 */
object UserInfoApp extends BaseApp {
  override var appName: String = "UserInfoApp"
  override var duration: Int = 10

  def main(args: Array[String]): Unit = {

    run{

      val ds: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Constansts.GMALL_USER_INFO, streamingContext)

      ds.foreachRDD(rdd =>
        rdd.foreachPartition(partition => {

          //获取连接
          val jedis: Jedis = RedisUtil.getJedisClient()

          partition.foreach(record => {

            val key: String = JSON.parseObject(record.value()).getString("id")

            jedis.set("user_id:"+key, record.value())

          })

          jedis.close()

        })

      )

    }

  }
}
